package net.j4love.nio.channels;

import org.junit.Assert;
import org.junit.Test;

import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.NetworkInterface;
import java.net.StandardProtocolFamily;
import java.net.StandardSocketOptions;
import java.net.UnknownHostException;
import java.nio.ByteBuffer;
import java.nio.channels.DatagramChannel;
import java.nio.channels.MembershipKey;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.charset.Charset;
import java.util.Enumeration;
import java.util.Iterator;
import java.util.Objects;
import java.util.Set;

/**
 * @author he peng
 * @create 2018/6/7 16:39
 * @see
 */
public class MulticastChannelTest {

    static final int JVM_BUG_THRESHOLD = 5;
    final InetAddress multicastAddress = InetAddress.getByName("224.1.2.3");
    final int multicastBindPort0 = 5000;
    final int multicastBindPort1 = 5001;


    public MulticastChannelTest() throws UnknownHostException {
    }

    @Test
    public void multicastTest0() throws Exception {

        /*
         *  创建指定协议的组播通道,
         *  1.INET:IPV4
         *  2.INET6:IPV6
         */
        DatagramChannel dChannel = DatagramChannel.open(StandardProtocolFamily.INET);
        Assert.assertTrue(dChannel.isOpen());

        NetworkInterface networkInterface = NetworkInterface.getByName("eth0");

        Enumeration<NetworkInterface> networkInterfaces = NetworkInterface.getNetworkInterfaces();
        while (networkInterfaces.hasMoreElements()) {
            NetworkInterface ni = networkInterfaces.nextElement();
            if (ni.isVirtual()) {
                continue;
            }
            if (ni.supportsMulticast()) {
                networkInterface = ni;
                break;
            }
        }

         /*
         *  setOption(StandardSocketOptions.SO_REUSEADDR, true)
         *  表示允许组播成员绑定到相同的端口上，它必须在绑定bind()前调用。
         *  由于bind()方法内的参数绑定的是当前组成员用于接收数据报的本地端口，
         *  因此如果此终端只用于发送，则将bind()方法的参数设置为null或直接去掉此方法。
         */
        dChannel.setOption(StandardSocketOptions.SO_REUSEADDR , true);

        dChannel.bind(new InetSocketAddress(multicastBindPort0));

        // 允许接收自己发送出去的数据报
        dChannel.setOption(StandardSocketOptions.IP_MULTICAST_IF , networkInterface);
        MembershipKey membershipKey = dChannel.join(multicastAddress, networkInterface);
        System.out.println("[" + dChannel.getLocalAddress() + "] join multicast group [" + multicastAddress + "]");

        dChannel.configureBlocking(false);

        Selector selector = Selector.open();
        dChannel.register(selector , SelectionKey.OP_READ);
        int jvmBug = 0;

        while (true) {
            System.out.println("*************************************");
            Set<SelectionKey> keys = selector.keys();
            int selectedNum = selector.select();

            if (selectedNum == 0) {
                // 解决 nio 空轮训 bug
                jvmBug++;
                if (jvmBug > JVM_BUG_THRESHOLD) {
                    Selector newSelector = Selector.open();

                    // 这行代码在请求量大的情况下会产生 java.util.ConcurrentModificationException 异常 ， 没搞清楚原因是什么
                    for (SelectionKey sk : selector.keys()) {
                        if (!sk.isValid() || sk.interestOps() == 0) {
                            continue;
                        }
                        SelectableChannel channel = sk.channel();
                        if (Objects.nonNull(channel) && channel.isOpen()) {
                            channel.register(newSelector, sk.interestOps());
                        }
                    }

                    Selector oldSelector = selector;
                    selector = newSelector;
                    if (oldSelector.isOpen()) {
                        oldSelector.close();
                    }

                    System.out.println("[" + Thread.currentThread().getName() + "]" + "Fix Nio epoll empty polling bug .......");
                    jvmBug = 0;
                }
                continue;
            }

            Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
            while (iterator.hasNext()) {
                SelectionKey sk = iterator.next();
                if (! sk.isValid()) {
                    sk.cancel();
                    continue;
                } else if (sk.isAcceptable()) {
                    System.out.println("Accept ready");
                } else if (sk.isConnectable()) {
                    System.out.println("Connect ready");
                } else if (sk.isReadable()) {
                    System.out.println("Read ready");
                    DatagramChannel channel = (DatagramChannel) sk.channel();
                    if (Objects.isNull(channel) || ! channel.isOpen()) {
                        channel.close();
                        System.out.println("channel close");
                        continue;
                    }
                    ByteBuffer readBuf = ByteBuffer.allocateDirect(1024);
                    channel.receive(readBuf);
                    readBuf.flip();

                    byte[] readBytes = new byte[readBuf.limit()];
                    readBuf.get(readBytes);
                    System.out.println("Receive from [" + channel.getRemoteAddress() + "] message -> " + new String(readBytes, Charset.forName("utf-8")));
                    String sendMessage = "hello everyone , i am Xiao Ming";
                    byte[] sendBytes = sendMessage.getBytes(Charset.forName("utf-8"));
                    ByteBuffer sendBuf = ByteBuffer.allocateDirect(sendBytes.length);
                    sendBuf.put(sendBytes);
                    sendBuf.flip();
                    channel.send(sendBuf , new InetSocketAddress(multicastAddress , multicastBindPort1));
                    System.out.println("write to [" + channel.getRemoteAddress() + "] -> message -> " + sendMessage);
                } else if (sk.isWritable()) {
                    System.out.println("Write ready");

                } else {
                    System.out.println("Unknown SelectionKey");
                }

                iterator.remove();

            }
        }
    }

    @Test
    public void multicastTest1() throws Exception {

        /*
         *  创建指定协议的组播通道,
         *  1.INET:IPV4
         *  2.INET6:IPV6
         */
        DatagramChannel dChannel = DatagramChannel.open(StandardProtocolFamily.INET);
        Assert.assertTrue(dChannel.isOpen());

        NetworkInterface networkInterface = NetworkInterface.getByName("eth0");

        Enumeration<NetworkInterface> networkInterfaces = NetworkInterface.getNetworkInterfaces();
        while (networkInterfaces.hasMoreElements()) {
            NetworkInterface ni = networkInterfaces.nextElement();
            if (ni.isVirtual()) {
                continue;
            }
            if (ni.supportsMulticast()) {
                networkInterface = ni;
                break;
            }


        }

         /*
         *  setOption(StandardSocketOptions.SO_REUSEADDR, true)
         *  表示允许组播成员绑定到相同的端口上，它必须在绑定bind()前调用。
         *  由于bind()方法内的参数绑定的是当前组成员用于接收数据报的本地端口，
         *  因此如果此终端只用于发送，则将bind()方法的参数设置为null或直接去掉此方法。
         */
        dChannel.setOption(StandardSocketOptions.SO_REUSEADDR , true);

        dChannel.bind(new InetSocketAddress(multicastBindPort1));

        // 允许接收自己发送出去的数据报
        dChannel.setOption(StandardSocketOptions.IP_MULTICAST_IF , networkInterface);
        MembershipKey membershipKey = dChannel.join(multicastAddress, networkInterface);
        System.out.println("[" + dChannel.getLocalAddress() + "] join multicast group [" + multicastAddress + "]");

        dChannel.configureBlocking(false);

        Selector selector = Selector.open();
        dChannel.register(selector , SelectionKey.OP_WRITE);

        int jvmBug = 0;
        while (true) {
            int selectedNum = selector.select();

            if (selectedNum == 0) {
                // 解决 nio 空轮训 bug
                jvmBug++;
                if (jvmBug > JVM_BUG_THRESHOLD) {
                    Selector newSelector = Selector.open();

                    // 这行代码在请求量大的情况下会产生 java.util.ConcurrentModificationException 异常 ， 没搞清楚原因是什么
                    for (SelectionKey sk : selector.keys()) {
                        if (!sk.isValid() || sk.interestOps() == 0) {
                            continue;
                        }
                        SelectableChannel channel = sk.channel();
                        if (Objects.nonNull(channel) && channel.isOpen()) {
                            channel.register(newSelector, sk.interestOps());
                        }
                    }

                    Selector oldSelector = selector;
                    selector = newSelector;
                    if (oldSelector.isOpen()) {
                        oldSelector.close();
                    }

                    System.out.println("[" + Thread.currentThread().getName() + "]" + "Fix Nio epoll empty polling bug .......");
                }
                jvmBug = 0;
                continue;
            }

            Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
            while (iterator.hasNext()) {
                SelectionKey sk = iterator.next();
                if (! sk.isValid()) {
                    sk.cancel();
                    continue;
                } else if (sk.isAcceptable()) {
                    System.out.println("Accept ready");
                } else if (sk.isConnectable()) {
                    System.out.println("Connect ready");
                } else if (sk.isReadable()) {
                    System.out.println("Read ready");
                    DatagramChannel channel = (DatagramChannel) sk.channel();
                    if (Objects.isNull(channel) || ! channel.isOpen()) {
                        channel.close();
                        System.out.println("channel close");
                        continue;
                    }
                    ByteBuffer readBuf = ByteBuffer.allocateDirect(1024);
                    channel.receive(readBuf);
                    readBuf.flip();

                    byte[] readBytes = new byte[readBuf.limit()];
                    readBuf.get(readBytes);
                    System.out.println("Receive from [" + channel.getRemoteAddress() + "] message -> " + new String(readBytes, Charset.forName("utf-8")));
                    String sendMessage = "hello everyone , i am Danny";
                    byte[] sendBytes = sendMessage.getBytes(Charset.forName("utf-8"));
                    ByteBuffer sendBuf = ByteBuffer.allocateDirect(sendBytes.length);
                    sendBuf.put(sendBytes);
                    sendBuf.flip();
                    channel.send(sendBuf , new InetSocketAddress(multicastAddress , multicastBindPort0));
                    System.out.println("write to [" + channel.getRemoteAddress() + "] -> message -> " + sendMessage);
                } else if (sk.isWritable()) {
                    System.out.println("Write ready");
                    DatagramChannel channel = (DatagramChannel) sk.channel();
                    if (Objects.isNull(channel) || ! channel.isOpen()) {
                        channel.close();
                        continue;
                    }
                    String sendMessage = "hello everyone , i am Danny";
                    byte[] sendBytes = sendMessage.getBytes(Charset.forName("utf-8"));
                    ByteBuffer sendBuf = ByteBuffer.allocateDirect(sendBytes.length);
                    sendBuf.put(sendBytes);
                    sendBuf.flip();
                    channel.send(sendBuf , new InetSocketAddress(multicastAddress , multicastBindPort0));
                    System.out.println("write to [" + channel.getRemoteAddress() + "] -> message -> " + sendMessage);
                    channel.register(selector , SelectionKey.OP_READ);
                } else {
                    System.out.println("Unknown SelectionKey");
                }

                iterator.remove();
            }
        }
    }

}
